package com.flute.haflute.net.messaging;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.service.IoHandler;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.DatagramSessionConfig;
import org.apache.mina.transport.socket.nio.NioDatagramAcceptor;
import org.apache.mina.transport.socket.nio.NioDatagramConnector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.agent.AgentContext;
import com.flute.haflute.agent.heartbeat.HeartBeatListener4Master;
import com.flute.haflute.agent.job.JobListener4Cluster;
import com.flute.haflute.agent.job.JobListener4Node;
import com.flute.haflute.agent.manage.ClusterManagementListener;
import com.flute.haflute.agent.topo.ClusterMessageListener;
import com.flute.haflute.common.ClusterConstants.ROLE;
import com.flute.haflute.common.ClusterContext;
import com.flute.haflute.jobbox.base.RunProcessInterface;
import com.flute.haflute.net.HafluteKeepAliveFilterInMina;
import com.flute.haflute.tools.Utils;

/**
 * 集群的消息服务中心，负责实现网络端口的监听、连接和消息流发送服务<br>
 * 
 */
public class MessagingService implements RunProcessInterface {

	private static Logger logger = LoggerFactory
			.getLogger(MessagingService.class);

	private static MessagingService instance = new MessagingService();

	/** messageSubscribers 所有订阅消息的消费者，以<动词，订阅者>对形式出现 */
	private Map<Message.Verb, ClusterMessageListener> messageSubscribers;

	private NioDatagramAcceptor acceptor;

	/**
	 * 返回单例对象的实例
	 * 
	 * @return instance MessagingService的实例
	 */
	public static MessagingService getInstance() {
		return instance;
	}

	/**
	 * @throws IOException
	 */
	private MessagingService() {
		logger.info("initialized Messaging Service OK");
	}

	private void startup() throws IOException {
		messageSubscribers = new HashMap<Message.Verb, ClusterMessageListener>();
		messageSubscribers.put(Message.Verb.JOB_SCHEDULE,
				new JobListener4Node());
		messageSubscribers.put(Message.Verb.CLUSTER_MANAGE,
				new ClusterManagementListener());

		if (AgentContext.getNodeRole() == ROLE.DM) {
			messageSubscribers.put(Message.Verb.HEARTBEARTING,
					new HeartBeatListener4Master());
			messageSubscribers.put(Message.Verb.JOB_NEW,
					new JobListener4Cluster());
		}
		listen();
	}

	@Override
	public void shutdown() {
		if (acceptor != null && acceptor.isActive()) {
			/**
			 * 经测试 acceptor使用dispose无法停止
			 */
			// acceptor.dispose();
		}
	}

	/**
	 * 监听集群端口，以便和其他节点通信
	 * 
	 * @throws IOException
	 */
	private void listen() throws IOException {
		acceptor = new NioDatagramAcceptor();
		acceptor.setHandler(new MessageHandlerInMina(messageSubscribers));

		DefaultIoFilterChainBuilder chain = acceptor.getFilterChain();
		chain.addLast("keep-alive", new HafluteKeepAliveFilterInMina());
		chain.addLast("toMessageTyep", new SinkFilterInMina());

		if (logger.isDebugEnabled()) {
			chain.addLast("logger", new LoggingFilter());
		}

		DatagramSessionConfig dcfg = acceptor.getSessionConfig();
		dcfg.setReuseAddress(true);

		acceptor.bind(new InetSocketAddress(ClusterContext.getHeartBeatPort()));
		logger.info("Cluster Agent is now listening on port:{}",
				ClusterContext.getHeartBeatPort());
	}

	/**
	 * 
	 * 连接以后，会有一个后台线程负责等待session关闭后清理资源
	 * 
	 * @param nodeAddress
	 * @param handler
	 * @return IoSession
	 * 
	 */
	public IoSession connect2Node(InetSocketAddress nodeAddress,
			IoHandler handler) {
		final NioDatagramConnector connector = new NioDatagramConnector();
		connector.setConnectTimeoutMillis(60000L);
		connector.setConnectTimeoutCheckInterval(10000);
		connector.setHandler(handler);

		DefaultIoFilterChainBuilder chain = connector.getFilterChain();
		chain.addLast("keep-alive", new HafluteKeepAliveFilterInMina());
		chain.addLast("toMessageTyep", new SinkFilterInMina());

		if (logger.isDebugEnabled()) {
			chain.addLast("logger", new LoggingFilter());
		}

		ConnectFuture connFuture = connector.connect(nodeAddress);
		connFuture.awaitUninterruptibly();
		final IoSession session = connFuture.getSession();
		Thread cleanerThread = new Thread() {
			@Override
			public void run() {
				session.getCloseFuture().awaitUninterruptibly();
				connector.dispose();
			}
		};
		cleanerThread.setName("MINA-DISPOSER_" + new Date());
		cleanerThread.start();
		return session;
	}

	/**
	 * 通过MINA框架发送Message消息的方法<br>
	 * 
	 * 有两种方法处理session的关闭：
	 * 
	 * 1. 在IoHandler中处理，比如数据通信完毕就关闭 ;
	 * 
	 * 2. 通过得到返回的结果Session，控制通信的关闭。
	 * 
	 * @param nodeAddress
	 * @param handler
	 * @return IoSession
	 * 
	 * @throws IOException
	 */
	public IoSession send2Node(InetSocketAddress nodeAddress, Message message,
			IoHandler handler) throws IOException {
		IoSession session = connect2Node(nodeAddress, handler);
		Utils.sendMessage(session, message);
		return session;
	}

	@Override
	public void callMain(String... params) throws Exception {
		startup();
	}

	@Override
	public boolean isInitialized() {
		if (acceptor != null) {
			return true;
		}
		return false;
	}

	@Override
	public void restart() {
	}

}